//package com.sunye.kafka2;
//
//import org.apache.kafka.clients.producer.Partitioner;
//import org.apache.kafka.clients.producer.internals.StickyPartitionCache;
//import org.apache.kafka.common.Cluster;
//import org.apache.kafka.common.PartitionInfo;
//import org.apache.kafka.common.utils.Utils;
//
//import java.util.List;
//import java.util.Map;
//
///**
// * 生产者自定义分区策率
// */
//public class CustomizePartitioner implements Partitioner {
//    private final StickyPartitionCache stickyPartitionCache = new StickyPartitionCache();
//
//    @Override
//    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
//        if (keyBytes == null) {
//            return stickyPartitionCache.partition(topic, cluster);
//        }
//        if("sunyeKey".equals(key)){
//            return 5;//如果传入的key是sunyeKey放入到5分区
//        }
//        List<PartitionInfo> partitionInfos = cluster.partitionsForTopic(topic);
//        int numPartitions = partitionInfos.size();
//        // hash the keyBytes to choose a partition
//        return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
//    }
//
//    @Override
//    public void close() {
//
//    }
//
//    @Override
//    public void configure(Map<String, ?> configs) {
//
//    }
//}
